package edu.ncsu.csc.streamingmr;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;

public class FixedLengthTextInputStreamFormat
		implements InputStreamFormat<FixedLengthText, LongWritable> {
	
	private int batchsize;
	private static final int recordsize=FixedLengthText.getLength()+8; //8 bytes are the length of a long
	
	public int getBatchsize(){
		return this.batchsize;
	}
	
	
	public FixedLengthTextInputStreamFormat(JobConf conf) {
		batchsize=conf.getInt("mapred.inputstream.batchsize", 100*1024);
	}
	
	public class FixedLengthTextRecordReader 
		implements RecordReader<FixedLengthText, LongWritable> {

		private ByteBuffer batchbytes;
		
		public FixedLengthTextRecordReader(byte[] batchbytes) {
			this.batchbytes=ByteBuffer.wrap(batchbytes);
			//System.out.print("testfji in FixedLengthTextRecordReader: bach= "+new String(batchbytes)
				//+"with length= "+batchbytes.length+" limit="+this.batchbytes.limit());
		}
		@Override
		public void close() throws IOException {
			
		}

		@Override
		public FixedLengthText createKey() {
			return new FixedLengthText();
		}

		@Override
		public LongWritable createValue() {
			return new LongWritable();
		}

		@Override
		public long getPos() throws IOException {
			return batchbytes.position();
		}

		@Override
		public float getProgress() throws IOException {
			return (float)getPos()/(float)batchbytes.limit();
		}

		@Override
		public boolean next(FixedLengthText key, LongWritable value)
				throws IOException {
			byte[] keytext=new byte[FixedLengthText.getLength()];
			byte[] valuebytes= new byte[8]; //8 is the size of a long
			DataInput valuein=new DataInputStream(new ByteArrayInputStream(valuebytes));
			//System.out.println("testfji enterd next(): getPos="+getPos());
			if(getPos()+recordsize<=batchbytes.limit()) {
				batchbytes.get(keytext);
				key.set(keytext);
				batchbytes.get(valuebytes);
				value.readFields(valuein);
				return true;
			}
			return false;
		}
		
	}
	@Override
	public RecordReader<FixedLengthText, LongWritable> getRecordReader(StreamBatch batch) {
		byte[] streambatch=batch.getByteArray();
		return new FixedLengthTextRecordReader(streambatch);		
	}
	
	int[] counts = new int[26];

	private byte[] leftoverbuffer=null;
	@Override
	public StreamBatch[] getStreamBatch(byte[] inputstream) throws IOException{
		
		//System.out.println("entered getStreamBatch: ");
		//for(int i=0;i<inputstream.length;i++) {
			//System.out.print((char)inputstream[i]);
		//}
		//System.out.println();
		List<StreamBatch> batchlist=new ArrayList<StreamBatch>();
		if(leftoverbuffer!=null) {
			ByteBuffer inputbuf=ByteBuffer.allocate(leftoverbuffer.length
						+inputstream.length);
			inputbuf.put(leftoverbuffer);
			inputbuf.put(inputstream);
			inputstream=inputbuf.array();
			leftoverbuffer=null;
		}
		
		int pos=0, batchsize=0;
		ByteBuffer thebatch=null;
		int wordstart=0;
		boolean inword=false;
		FixedLengthText text=new FixedLengthText();
		ByteArrayOutputStream longstream=new ByteArrayOutputStream();
		DataOutputStream longout=new DataOutputStream(longstream);
		
		while(pos<inputstream.length) {
			byte c=inputstream[pos];
			if(!((c>='a' && c<='z')||(c>='A' && c<='Z')||(c>='0' && c<='9'))) {
				if(inword) { //inword==true, encounters the end of a word
					
					//write out the FixedLengthText
					text.set(inputstream, wordstart, Math.min(pos-wordstart, FixedLengthText.getLength()));
					//System.out.println("testfji text has length"+text.getText().length+" thebatch has "+thebatch.remaining());
					//for(int i=0;i<text.getText().length;i++)
						//System.out.print((char)text.getText()[i]);
					//System.out.println("  ENDtestfji\n.");
					thebatch.put(text.getText()); //fixed length text written into the buffer
					/*counts[text.getText()[0]-97]++;
					System.out.println("counts[" + (text.getText()[0]-97) + "] = " + counts[text.getText()[0]-97]);*/ 
					

					
					//write out the long
					longstream.reset();
					new LongWritable(1).write(longout);
					thebatch.put(longstream.toByteArray());
					
					inword=false;
					batchsize+=(recordsize);
					
					
					if(batchsize+recordsize>this.batchsize) {
						thebatch.flip();
						byte batchbytes[]=new byte[thebatch.limit()];
						thebatch.get(batchbytes);
						batchlist.add(new StreamBatch(batchbytes));
						batchsize=0;
						thebatch=ByteBuffer.allocate(this.batchsize);
					}
				}
			} else { // encounters non-delimiter character
				if(!inword) {
					if(thebatch==null) {
						thebatch=ByteBuffer.allocate(this.batchsize);
						batchsize=0;
					}
					inword=true;
					wordstart=pos;
				}
				
			}
			pos++;
		}
		if(batchsize!=0) {
			if(inword) {//flush the final word to leftoverbuffer
				this.leftoverbuffer=new byte[pos-wordstart];
				for(int i=0;i+wordstart<pos;i++)
					this.leftoverbuffer[i]=inputstream[wordstart+i];	
			}
			//write out the last batch
			thebatch.flip();
			byte batchbytes[]=new byte[thebatch.limit()];
			thebatch.get(batchbytes);
			batchlist.add(new StreamBatch(batchbytes));
		}
		
		
		StreamBatch[] batches=new StreamBatch[batchlist.size()];
		batchlist.toArray(batches);
		return batches;
	}


	@Override
	public int getRecordSize() {
		return recordsize;
	}

}
